package asynctask

import (
	"encoding/json"
	"fmt"
	"scase.io/application-auto-scaling-service/pkg/api/model/cci"
	"scase.io/application-auto-scaling-service/pkg/cloudresource"
	"scase.io/application-auto-scaling-service/pkg/common"
	"scase.io/application-auto-scaling-service/pkg/db"
	event "scase.io/application-auto-scaling-service/pkg/service/event"
	"scase.io/application-auto-scaling-service/pkg/utils/logger"
	"sync"
	"time"
)

const (
	WaitPodCreateTimes = 10
)

// ScaleOutTask 扩容任务
type PodScaleOutTask struct {
	TaskId      string
	GroupId     string
	InstanceNum int32
	BaseTask
	event *event.EventService
}

// NewScaleOutTask ...
func NewPodScaleOutTask(taskId string, groupId string) *PodScaleOutTask {
	return &PodScaleOutTask{
		TaskId:  taskId,
		GroupId: groupId,
	}
}

// GetKey get id of the resource corresponding to the task
func (t *PodScaleOutTask) GetKey() string {
	return t.GroupId
}

// GetType get task type
func (t *PodScaleOutTask) GetType() string {
	return db.TaskTypeScaleOutPodGroup
}

func (t *PodScaleOutTask) Run(log *logger.FMLogger) error {
	// 1. 初始化任务
	if err := t.initTask(log); err != nil {
		log.Error("[pod_scale_out] init task failed with error:%+v", err)
		return db.TxRecordPodGroupScalingTaskComplete(t.GroupId, t.TaskId)
	}
	group, err := db.ScalingGroupTable().GetScalingGroupById(t.GroupId)
	if err != nil {
		log.Error("[pod_scale_out] get pod scaling group [%s] error:%+v", t.GroupId, err)
		return err
	}
	podCtrl, err := cloudresource.GetPodController(group.ProjectId)
	if err != nil {
		log.Error("[pod_scale_out] get controller of group [%s] error:%+v", t.GroupId, err)
		return err
	}
	t.event, err = event.NewEventService(group.ProjectId, nil, log)
	if err != nil {
		log.Error("[pod_scale_out] new event service error:%+v", err)
		return err
	}
	go t.event.BuildAndRegisterEvent(fmt.Sprintf("[Scale-out-%d-pod] start", t.InstanceNum), group.Id, fmt.Sprintf("fleet [%s] scale out %d pod start", group.FleetId, t.InstanceNum),
		fmt.Sprintf("fleet [%s] scale out %d instance start", group.FleetId, t.InstanceNum), common.EventTraceStateNormal)
	// 2. 检查namespace是否存在
	if err := podCtrl.QueryNamespaceExist(group.FleetId); err != nil {
		log.Error("[pod_scale_out] get namespace of [%s] error:%+v", group.FleetId, err)
		go t.event.BuildAndRegisterEvent(fmt.Sprintf("[Scale-out-%d-pod] failed", t.InstanceNum), group.Id, fmt.Sprintf("create pod task [%s] failed with error:%+v", t.TaskId, err.Error()),
			fmt.Sprintf("fleet [%s] scale out %d instance failed", group.FleetId, t.InstanceNum), common.EventTraceStateIncident)
		return err
	}

	// 3. 创建pod
	if err := t.CreatePod(log, group, podCtrl); err != nil {
		log.Error("[pod_scale_out] create pod error:%+v", err)
		go t.event.BuildAndRegisterEvent(fmt.Sprintf("[Scale-out-%d-pod] failed", t.InstanceNum), group.Id, fmt.Sprintf("create pod task [%s] failed with error:%+v", t.TaskId, err.Error()),
			fmt.Sprintf("fleet [%s] scale out %d instance failed", group.FleetId, t.InstanceNum), common.EventTraceStateIncident)
		return err
	}
	go t.event.BuildAndRegisterEvent(fmt.Sprintf("[Scale-out-%d-pod] success", t.InstanceNum), group.Id, fmt.Sprintf("fleet [%s] scale out %d pod success", group.FleetId, t.InstanceNum),
		fmt.Sprintf("fleet [%s] scale out %d instance success", group.FleetId, t.InstanceNum), common.EventTraceStateNormal)

	return db.TxRecordPodGroupScalingTaskComplete(t.GroupId, t.TaskId)
}

func (t *PodScaleOutTask) initTask(log *logger.FMLogger) error {
	// init task
	taskInfo, err := db.GetAsyncTaskByTaskId(t.TaskId)
	if err != nil {
		log.Error("[pod_scale_out] get task [%s] error:%+v", err)
		return err
	}
	if taskInfo.AffectNum <= 0 {
		log.Error("[pod_scale_out] task [%s] will create 0 instance, quit task", t.TaskId)
		return fmt.Errorf("[pod_scale_out] task [%s] will create 0 instance, quit task", t.TaskId)
	}
	t.InstanceNum = taskInfo.AffectNum
	taskRetryTimes := t.GetRetryTimes()
	if taskRetryTimes > common.AsyncScaleOutMaxRetryTimes {
		// 任务下发失败时修改期望实例数
		UpdateDesireNum(log, t.GroupId, t.InstanceNum)
		log.Error("[pod_scale_out] task [%s] execute fail after retry %d times, give up on this task", t.TaskId, taskRetryTimes)
		return fmt.Errorf("[pod_scale_out] task [%s] execute fail after retry %d times, give up on this task", t.TaskId, taskRetryTimes)
	}
	log.Info("async task [%s] group [%s] scale out %d start", t.TaskId, t.GroupId, t.InstanceNum)
	return nil
}

func (t *PodScaleOutTask) CreatePod(log *logger.FMLogger, group *db.ScalingGroup, podCtrl *cloudresource.PodResourceController) error {
	instanceNum := int(t.InstanceNum)
	// 1. 创建的pod只有name不一样，name作为唯一索引， 数据库存入初始状态
	nameList, requestList, err := t.buildRequest(group, log)
	// 2. 启动goroutine前，将准备创建的存入数据库，以便重入
	t.saveToDB(log, nameList, group)

	// 3. 启动goroutine
	if err != nil {
		log.Error("[pod_scale_out] build request list with error:%+v", err)
		return err
	}
	wg := sync.WaitGroup{}
	wg.Add(instanceNum)
	for i := 0; i < instanceNum; i++ {
		if i > 0 && i%common.PodCountToSleep == 0 {
			time.Sleep(time.Minute * common.SleepMinToPod)
		}
		index := i
		go func(index int) {
			defer wg.Done()
			if err := podCtrl.CreatePod(requestList[index], group.FleetId); err != nil {
				log.Error("[pod_scale_out] create pod error:%+v", err)
				return
			}
			if err := t.waitPodCreate(log, group.FleetId, nameList[index], podCtrl, group); err != nil {
				log.Error("[pod_scale_out] wait pod [%s] create error:%+v", nameList[index], err)
				return
			}
		}(index)
	}
	wg.Wait()
	// 4. 如果goroutine结束后，数据库中还有没创成功的，视为失败，删除相应的pod，等待下次重入
	if err := t.handleFailJob(log, podCtrl); err != nil {
		log.Error("[pod_scale_out] handle fail jobs failed or pod create failed, error:%+v", err)
		return err
	}
	// 5. 更新当前实例数
	if err := t.updateCurrentNum(log); err != nil {
		log.Error("[pod_scale_out] update current instance num of group [%s] error:%+v", t.GroupId, err)
		return err
	}
	return nil
}

func (t *PodScaleOutTask) buildRequest(group *db.ScalingGroup, log *logger.FMLogger) ([]string, []interface{}, error) {
	podreq := &cci.PodRequest{}
	if err := json.Unmarshal([]byte(group.PodTemplate), podreq); err != nil {
		log.Error("[pod_scale_out] unmarshal pod template of group [%s] error:%+v", t.GroupId, err)
		return nil, nil, err
	}
	reqList := make([]interface{}, t.InstanceNum)
	var privateBandWithReq *cci.PodPrivateBandwidthRequest = nil
	var shareBandWithReq *cci.PodSharedBandwidthRequest = nil
	// 判断带宽类型
	if len(podreq.Metadata.Annotations.YangtseIoEipBandwidthId) == 0 {
		// 独享带宽
		if err := json.Unmarshal([]byte(group.PodTemplate), &privateBandWithReq); err != nil {
			log.Error("[pod_scale_out] Unmarshal pod template {%s} error:%+v", group.PodTemplate, err)
			return nil, nil, err
		}
	} else {
		//共享带宽
		if err := json.Unmarshal([]byte(group.PodTemplate), &shareBandWithReq); err != nil {
			log.Error("[pod_scale_out] Unmarshal pod template {%s} error:%+v", group.PodTemplate, err)
			return nil, nil, err
		}
	}
	names := make([]string, t.InstanceNum)
	instanceNum := int(t.InstanceNum)
	for i := 0; i < instanceNum; i++ {
		names[i] = fmt.Sprintf("fleet-%s-%s", group.FleetId, common.RandStrLowerCase(8))
		if privateBandWithReq != nil {
			privateBandWithReq.Metadata.Name = names[i]
			tmp := *privateBandWithReq
			reqList[i] = tmp
		} else {
			shareBandWithReq.Metadata.Name = names[i]
			tmp := *shareBandWithReq
			reqList[i] = tmp
		}
	}
	return names, reqList, nil
}

func (t *PodScaleOutTask) saveToDB(log *logger.FMLogger, nameList []string, group *db.ScalingGroup) {
	wg := sync.WaitGroup{}
	wg.Add(len(nameList))
	// 注意：goroutine 传递的是引用
	for index, _ := range nameList {
		if index >= 0 && index%common.InstanceCountToSleep == 0 {
			time.Sleep(time.Second)
		}
		go func(index int) {
			defer wg.Done()
			podInfo := db.PodInfo{
				HostName:       nameList[index],
				FleetId:        group.FleetId,
				Namespace:      group.FleetId,
				ProjectId:      group.ProjectId,
				ScalingGroupId: group.Id,
				InstanceId:     nameList[index],
				State:          db.PodStatePending,
				TaskId:         t.TaskId,
				TimeModel: db.TimeModel{
					IsDeleted: "0",
				},
			}
			if err := db.PodInfoTable().Insert(&podInfo); err != nil {
				log.Error("[pod_scale_out] insert pod info to db error :%+v", err)
				return
			}
		}(index)
	}
	wg.Wait()
}

func (t *PodScaleOutTask) updateSuccessPod(log *logger.FMLogger, podname, publidIP, PodIP string) error {
	filter := db.Filters{"HostName": podname}
	podInfo, err := db.PodInfoTable().Get(filter)
	if err != nil {
		log.Error("[pod_scale_out] get pod info of [%s] error:%+v", podname, err)
		return err
	}
	podInfo.State = db.PodStateRunning
	podInfo.PrivateIp = PodIP
	podInfo.PublicIp = publidIP
	if err := db.PodInfoTable().Update(podInfo); err != nil {
		log.Error("[pod_scale_out] update pod info of {%+v} error:%+v", podInfo, err)
		return err
	}
	return nil
}

// 更新group的当前实例数
func (t *PodScaleOutTask) updateCurrentNum(log *logger.FMLogger) error {
	count, err := db.PodInfoTable().CalculatePodNumOfGroup(t.GroupId)
	if err != nil {
		log.Error("[pod_scale_out] get running pods of group [%s] error:%+v", t.GroupId, err)
		return err
	}
	if err := db.ScalingGroupTable().TxUpdateCurNum(t.GroupId, count); err != nil {
		log.Error("[pod_scale_out] update current instance num of group [%s] error:%+v", t.GroupId, err)
		return err
	}
	return nil
}

// 处理创建失败的pod，将未能从pending转running的pod置error
func (t *PodScaleOutTask) handleFailJob(log *logger.FMLogger, podCtrl *cloudresource.PodResourceController) error {
	count, failPods, err := db.PodInfoTable().GetFailJob(t.TaskId)
	if err != nil {
		log.Error("[pod_scale_out] get fail Pods failed, task [%s], group [%s], error:%+v", t.TaskId, t.GroupId, err)
		return err
	}
	if count == 0 {
		log.Info("[pod_scale_out] there are no fail job of task [%s] to handle", t.TaskId)
		return nil
	}
	wg := sync.WaitGroup{}
	wg.Add(count)
	for index, _ := range failPods {
		go func(index int) {
			defer wg.Done()
			// cci删除pod
			if err := podCtrl.DeletePod(failPods[index].FleetId, failPods[index].HostName); err != nil {
				log.Error("[pod_scale_out] delete pod [%s] error:%+v", failPods[index].HostName, err)
			}
			// 数据库将pod置delete
			if err := db.PodInfoTable().TxUpdateState(failPods[index].HostName, db.PodStateDeleted); err != nil {
				log.Error("[pod_scale_out] delete pod [%s] error:%+v", failPods[index].HostName, err)
			}
		}(index)
	}
	wg.Wait()
	return fmt.Errorf("there are %d pods create error:%+v", count, err)
}

func (t *PodScaleOutTask) waitPodCreate(log *logger.FMLogger, namespace, podName string, podCtrl *cloudresource.PodResourceController, group *db.ScalingGroup) error {
	for retry := 0; retry < WaitPodCreateTimes; retry++ {
		queryResp, err := podCtrl.QueryPod(namespace, podName)
		if err != nil {
			log.Error("[pod_scale_out] query pod error:%+v", err)
			return err
		}
		if queryResp.Status.ContainerStatuses == nil {
			log.Info("waiting pod [%s] create, container state is nil, retry %d times", podName, retry)
			time.Sleep(sleepSecond * time.Second)
			continue
		}
		// 创建成功后入库
		if queryResp.Status.ContainerStatuses[0].Ready == true && len(queryResp.Status.PodIP) != 0 {
			log.Info("[pod_scale_out] pod [%s] create success", podName)
			return t.updateSuccessPod(log, podName, queryResp.MetaData.Annotations.YangtseIoAllocatedIpv4Eip, queryResp.Status.PodIP)
		} else {
			log.Info("waiting pod [%s] create, container is not ready, retry %d times", podName, retry)
			time.Sleep(sleepSecond * time.Second)
			continue
		}
	}
	// 多次等待未返回就绪状态视为失败
	log.Error("create pod [%s] in namespace [%s] failed after %d retry", namespace, podName, WaitPodCreateTimes)
	return fmt.Errorf("create pod [%s] in namespace [%s] failed after %d retry", namespace, podName, WaitPodCreateTimes)
}
